package demo02;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class UsersAndMovies extends Configured implements Tool {

    public static Configuration getMyConfiguration() {
        //��������
        Configuration conf = new Configuration();
        //ͨ������ fs.hdfs.impl.disable.cache=true���û��棬ȷ��ÿ�� FileSystem.get()������ʵ��
        conf.setBoolean("fs.hdfs.impl.disable.cache", true);

        conf.setBoolean("mapreduce.app-submission.cross-platform", true);
        conf.set("dfs.client.socket-timeout", String.valueOf(10 * 60 * 1000)); // ���ó�ʱΪ60��
        conf.set("fs.defaultFS", "hdfs://192.40.10.130:9000");

        //**mapred-site.xml**
        //��windows��������������ԣ��������ô������
//        conf.set("mapreduce.framework.name","yarn"); // ָ��ʹ��yarn��� local

//		conf.set("mapreduce.job.jar",JarUtil.jar(UsersAndMovies.class));
        return conf;
    }

    @Override
    public int run(String[] args) throws Exception {
        if (args.length != 4) {
            System.err.println("demo01.UsersAndMovies <cachePath> <input> <output> <splitter>");
            System.exit(-1);
        }
        Configuration conf = UsersAndMovies.getMyConfiguration();

        //TODO:�޷���ȡ�����ļ�
//        DistributedCache.addCacheFile(new Path(args[0]).toUri(), conf);
        UsersMoviesMapper.movieFile = args[0];

        conf.set("SPLITTER", args[3]);
        Job job = Job.getInstance(conf, "joindata");
        job.setJarByClass(UsersAndMovies.class);
        job.setMapperClass(UsersMoviesMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setNumReduceTasks(0);
        FileInputFormat.addInputPath(job, new Path(args[1]));
        FileSystem.get(conf).delete(new Path(args[2]), true);
        FileOutputFormat.setOutputPath(job, new Path(args[2]));

        return job.waitForCompletion(true) ? -1 : 1;
    }

    public static void main(String[] args) {
        String[] myArgs = {
                "/movie/movies.dat",
                "/movie/ratings_users/part-m-00000",
                "/movie/users_movies",
                "::"
        };
        try {
            ToolRunner.run(getMyConfiguration(), new UsersAndMovies(), myArgs);
        } catch (Exception e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
    }

}
